package com.yanggu.flink.datastream_api.window

import cn.hutool.core.date.DateUtil
import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.configuration.{Configuration, RestOptions}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date

/**
 * 会话窗口(SessionWindow)
 * 由一系列事件组合一个指定时间长度的timeout间隙组成, 类似于web应用的session, 也就是说一段时间没有新的数据就会生成新的窗口
 * 特点: 时间无对齐
 * session窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况，相反，当它在一个固定的时间周期内不再收到元素，即非活动间隔产生，那个这个窗口就会关闭。
 * 一个session窗口通过一个session间隔来配置，这个session间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去。
 */
object SessionWindowDemo {

  def main(args: Array[String]): Unit = {
    //创建本地执行环境, 并且拥有WebUi和设置端口
    val config = new Configuration()
    config.setInteger(RestOptions.PORT.key(), 8080)
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
    environment.setParallelism(1)

    environment
      .socketTextStream("localhost", 9000)
      .map(value => {
        val arr = value.split(",")
        SensorReading(arr(0), arr(1).toLong * 1000, arr(2).toDouble)
      })
      //设置事件时间升序的Watermark
      .assignAscendingTimestamps(_.timestamp)
      //根据id进行分区
      .keyBy(_.id)
      //创建了事件时间会话窗口, 窗口大小为1minute
      .window(EventTimeSessionWindows.withGap(Time.minutes(1L)))
      //执行全窗口函数, 将窗口内的数据集收集齐后执行process方法
      .process(new ProcessWindowFunction[SensorReading, String, String, TimeWindow] {
        override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[String]): Unit = {
          val start = context.window.getStart
          val end = context.window.getEnd
          val result = s"当前时间: ${DateUtil.formatDateTime(new Date)}, " +
            s"会话窗口开始时间: ${DateUtil.formatDateTime(new Date(start))}, 窗口结束时间: ${DateUtil.formatDateTime(new Date(end))}, " +
            s"会话窗口的间隙: ${end - start}, 传感器id: $key, 总共有${elements.size}条传感器数据"
          out.collect(result)
        }
      })
      .print("result ")

    environment.execute("SessionWindowDemo Job")
  }

}
